package com.dylan.magic.rocketmq.config.mq;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;

/**
 * @author : ankang
 * @date : 2020/1/21 18:14
 * @Description : 消息生产者
 */
@Service
public class MessageProducer {
    private Logger logger = LoggerFactory.getLogger(MessageProducer.class);
    public static MessageProducer producer;

    private DefaultMQProducer defaultMqProducer;

    @PostConstruct
    public void init() throws MQClientException {
        //Instantiate with a producer group name.
        defaultMqProducer = new DefaultMQProducer(RocketMQConstants.PRODUCER_GROUP);
        // Specify name server addresses.
        defaultMqProducer.setNamesrvAddr(RocketMQConstants.NAME_SERVER_ADDRESS);
        //Launch the instance.
        defaultMqProducer.start();

        producer = this;
    }

    @PreDestroy
    public void preDestroy(){
        defaultMqProducer.shutdown();
    }

    public boolean send(MessageTopic topic, String keys, String content) {
        return send(topic, null, keys, content);
    }

    public boolean send(MessageTopic topic, String tags, String keys, String content) {
        boolean s = false;
        if (topic == null || keys == null || content == null) {
            logger.warn("MQ消息发送失败，参数错误，topic:{}，tags:{}，keys:{}，content:{}", topic, tags, keys, content);
            return s;
        }
        Message msg = new Message(topic.getTopic(), tags, keys, content.getBytes(StandardCharsets.UTF_8));
        try {
            SendResult sendResult = defaultMqProducer.send(msg);
            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                s = true;
                logger.info("MQ消息发送成功，message:{}，sendResult:{}", msg, sendResult);
            } else {
                logger.error("MQ消息发送失败，message:{}，sendResult:{}", msg, sendResult);
            }
        } catch (Exception e) {
            logger.error("MQ消息发送失败，message:" + msg, e);
        }
        return s;
    }
}
